package operator.function;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import util.Printer;
import util.ThreadHelper;

class RepeatWhenOperator {

    public static void main(String[] args) {
        repeatWhen1();
    }

    /**
     * 当Observable调用onComplete就会进入到repeatWhen方法里面，是否要触发重订阅
     * 就需要通过repeatWhen的Function方法所返回的ObservableSource确定
     * 如果返回的是onNext则触发重订阅，而返回的是onComplete/onError则不会触发重订阅
     */
    static void repeatWhen1(){
        Observable.just("A", "B").repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
                Printer.print(123);
                return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Object o) throws Exception {
                        Printer.print(456);
                        return Observable.timer(3000, TimeUnit.MILLISECONDS);
                    }
                });
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println(s);
            }
        });


        ThreadHelper.sleep();
    }


}
